package com.zeny.express.framework.mq.ibmmq;

import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ibm.mq.MQC;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.zeny.express.framework.mq.AbstractMqClient;
import com.zeny.express.framework.mq.MqCallback;
import com.zeny.express.framework.shared.serializer.JsonSerializer;

/**
 * 
 * @author KenQiu
 *
 * @param <T>
 */
@SuppressWarnings("unchecked")
public class IBMMqClient<T> extends AbstractMqClient<T> {
	protected final static Logger LOGGER = LoggerFactory
			.getLogger(IBMMqClient.class);

	protected IBMMqTemplate mqTemplate;
	protected MQGetMessageOptions gmo;
	protected MQPutMessageOptions pmo;

	private int gmoWaitInterval = 10000;

	public int getGmoWaitInterval() {
		return gmoWaitInterval;
	}

	public void setGmoWaitInterval(int gmoWaitInterval) {
		this.gmoWaitInterval = gmoWaitInterval;
	}

	public IBMMqClient() {
		this(true);
	}

	@SuppressWarnings("deprecation")
	public IBMMqClient(boolean syncpoint) {
		dataBinarySerializer = new JsonSerializer<T>(true, getDataClass());
		dataTextSerializer = new JsonSerializer<T>(false, getDataClass());

		mqTemplate = new IBMMqTemplate();

		gmo = new MQGetMessageOptions();
		gmo.options = MQC.MQGMO_WAIT + MQC.MQGMO_FAIL_IF_QUIESCING;

		pmo = new MQPutMessageOptions();

		if (syncpoint) {
			gmo.options += MQC.MQGMO_SYNCPOINT;
			pmo.options += MQC.MQPMO_SYNCPOINT;
		}
		gmo.waitInterval = gmoWaitInterval;
	}

	public T get() {
		return mqTemplate.execute(mqPool, new MqCallback<T, MQQueue>() {
			@Override
			public T doInMq(MQQueue queue) throws MQException, IOException {
				MQMessage message = new MQMessage();

				queue.get(message, gmo);

				int length = message.readInt();
				int dataFormat = message.readInt();
				byte[] bytes = new byte[length];
				message.readFully(bytes, 0, length);

				Object value = getDataSerializer(dataFormat).deserialize(bytes);
				return (T) value;
			}
		});
	}

	public void put(final T data) {
		final byte[] bytes = getDataSerializer(dataFormat).serialize(data);

		mqTemplate.execute(mqPool, new MqCallback<T, MQQueue>() {
			@Override
			public T doInMq(MQQueue queue) throws MQException, IOException {
				MQMessage message = new MQMessage();
				message.writeInt(bytes.length);
				message.writeInt(dataFormat);
				message.write(bytes);

				queue.put(message, pmo);

				return data;
			}
		});
	}
}
